//
//  rtmp_sender_impl.cpp
//  RtmpSender
//
//  Created by LSQ on 2020/2/12.
//  Copyright © 2020 Agora. All rights reserved.
//

#include "rtmp_sender_impl.h"

#include <unordered_map>

#include "facilities/environment/svc_env.h"
#include "modules/video_coding/codecs/parser/include/parser_h264.h"
#include "rtc_base/timeutils.h"
#include "third_party/srs-librtmp/src/srs/srs_librtmp.h"
#include "utils/log/log.h"
#include "utils/thread/base_worker.h"

namespace agora {
namespace rtc {

const char MODULE_NAME[] = "[RtmpSender]";
const int SRS_RECV_TIMEOUT_MS = 3000;
const int SRS_SEND_TIMEOUT_MS = 4500;

#define SRS_TO_AGORA_LOG(agoraFilter)                                        \
  do {                                                                       \
    char buf[256];                                                           \
    va_list args;                                                            \
    va_start(args, fmt);                                                     \
    vsnprintf(buf, sizeof(buf) - 1, fmt, args);                              \
    va_end(args);                                                            \
    log(agoraFilter, "%s[srs:%s:%d] %s", MODULE_NAME, tag, context_id, buf); \
  } while (0)

class SrsLogger : public ISrsLog {
 public:
  void verbose(const char* tag, int context_id, const char* fmt, ...) override {
    SRS_TO_AGORA_LOG(agora::commons::LOG_DEBUG);
  };

  void info(const char* tag, int context_id, const char* fmt, ...) override {
    SRS_TO_AGORA_LOG(agora::commons::LOG_DEBUG);
  }

  void trace(const char* tag, int context_id, const char* fmt, ...) override {
    SRS_TO_AGORA_LOG(agora::commons::LOG_INFO);
  }

  void warn(const char* tag, int context_id, const char* fmt, ...) override {
    SRS_TO_AGORA_LOG(agora::commons::LOG_WARN);
  }

  void error(const char* tag, int context_id, const char* fmt, ...) override {
    SRS_TO_AGORA_LOG(agora::commons::LOG_ERROR);
  }
};

RtmpSenderImpl::RtmpSenderImpl()
    : rtmp_handler_(nullptr),
      streaming_buffer_(nullptr),
      is_started_(false),
      is_stopping_(false),
      is_reconnecting_(false),
      reconnect_count_max_(10),
      reconnect_count_current_(0),
      last_video_width_(0),
      last_video_height_(0),
      sender_observers_(utils::RtcSyncCallback<IRtmpSenderObserver>::Create()),
      current_socket_state_(kSocketStateUnknown),
      is_sending_(false),
      send_frame_cnt_(0),
      relative_timestamps_(0) {}

RtmpSenderImpl::~RtmpSenderImpl() {
  log(commons::LOG_INFO, "%s %s dtor", MODULE_NAME, __FUNCTION__);
}

int RtmpSenderImpl::Initialize(const RtmpConfig& rtmp_config) {
  log(commons::LOG_INFO,
      "%s %s video(%dx%d@%d) audio(sample rate: %d, channels: %d,"
      " bytes per sample:%d)",
      MODULE_NAME, __FUNCTION__, rtmp_config.video_width, rtmp_config.video_height,
      rtmp_config.framerate, rtmp_config.audio_sample_rate_hz,
      rtmp_config.audio_numbder_of_channels, rtmp_config.audio_bytes_per_sample);
  rtmp_config_ = rtmp_config;
  rtmp_sender_worker_ = utils::minor_worker("RtmpSendWorker");
  streaming_buffer_ = std::make_unique<RtmpStreamingBuffer>(rtmp_sender_worker_);

  // initialize sending queue for sending audio/video packets
  rtmp_sending_queue_.reset(
      rtmp_sender_worker_->createAsyncQueue([](const task_type& task) { task(); }));
  rtmp_sending_queue_->set_priority(utils::IO_EVENT_PRIORITY_MISC);
  rtmp_sending_queue_->set_capacity(1000);

  // initialize srs log
  if (_srs_log != nullptr) {
    delete _srs_log;
  }
  _srs_log = new SrsLogger();
  return 0;
}

void RtmpSenderImpl::Release() {
  log(commons::LOG_INFO, "%s %s", MODULE_NAME, __FUNCTION__);
  rtmp_sending_queue_.reset();
  log(commons::LOG_INFO, "%s sending queue released", MODULE_NAME);

  // Uses sync call to flush the pending sender tasks.
  // Notes the minor worker may be shared by other components by design,
  // so simply resetting |rtmp_sender_worker_| may not flush sender tasks.
  rtmp_sender_worker_->sync_call(LOCATION_HERE, [this] {
    log(commons::LOG_INFO, "%s sender tasks flushed", MODULE_NAME);
    return 0;
  });

  streaming_buffer_.reset();
  if (reconnect_timer_) {
    reconnect_timer_.reset();
  }
  if (statistic_timer_) {
    statistic_timer_.reset();
  }
  if (last_statistic_data_) {
    last_statistic_data_.reset();
  }
  rtmp_sender_worker_.reset();

  if (_srs_log != nullptr) {
    delete _srs_log;
  }
  _srs_log = nullptr;

  delete this;
}

int RtmpSenderImpl::Start(const char* url) {
  log(commons::LOG_INFO, "%s %s, url: %s", MODULE_NAME, __FUNCTION__, url);

  if (!last_statistic_data_) {
    last_statistic_data_ = std::make_unique<STATISTIC_DATA>();
  }

  if (!statistic_timer_) {
    const uint64_t timer_interval_ms = 1000;
    statistic_timer_.reset(utils::major_worker()->createTimer(
        std::bind(&RtmpSenderImpl::StatisticTask, this), timer_interval_ms));
  }

  std::string tempUrl = url;
  return rtmp_sender_worker_->async_call(LOCATION_HERE, [this, tempUrl] {
    publish_url_ = tempUrl;
    CleanUpReconnection();
    // TODO(Haonong Yu): 2020/5/7 this task may be time consuming
    int ret = OpenRtmp();
    if (!ret) {
      is_started_ = true;
      is_stopping_ = false;

      send_frame_cnt_ = 0;
      streaming_buffer_->RegisterRtmpStreamingBufferObserver(this);
      UpdateStateAndNotifyIfNeeded(kSocketStateStartConnected);
      log(commons::LOG_INFO, "%s Open for write successfully", MODULE_NAME);
    } else if (ret == -ERROR_RTMP_STREAM_EXISTED) {
      UpdateStateAndNotifyIfNeeded(kSocketStateBadName);
      log(commons::LOG_ERROR, "%s Failed to open for write: %d", MODULE_NAME, ret);
    } else {
      UpdateStateAndNotifyIfNeeded(kSocketStateStartError);
      log(commons::LOG_ERROR, "%s Failed to open for write: %d", MODULE_NAME, ret);
    }
    // TODO(Haonong Yu): 2020/5/8 error code mapping
    return ret;
  });
}

void RtmpSenderImpl::Stop() {
  log(commons::LOG_INFO, "%s %s", MODULE_NAME, __FUNCTION__);

  if (statistic_timer_) {
    statistic_timer_.reset();
  }

  if (last_statistic_data_) {
    last_statistic_data_.reset();
  }

  // avoid socket receive timeout
  is_stopping_ = true;
  if (rtmp_handler_) {
    srs_rtmp_socket_shutdown(rtmp_handler_);
  }

  rtmp_sender_worker_->sync_call(LOCATION_HERE, [this] {
    if (!is_started_) {
      log(commons::LOG_WARN, "%s Stop: not started", MODULE_NAME);
      return -ERR_INVALID_STATE;
    }

    is_started_ = false;
    CleanUpReconnection();
    streaming_buffer_->CleanUp();
    streaming_buffer_->UnregisterRtmpStreamingBufferObserver(this);

    UpdateStateAndNotifyIfNeeded(kSocketStateStop);
    CloseRtmp();

    is_stopping_ = false;
    return -ERR_OK;
  });
}

int RtmpSenderImpl::OpenRtmp() {
  ASSERT_THREAD_IS(rtmp_sender_worker_->getThreadId());
  log(commons::LOG_INFO, "%s %s rtmp url: %s", MODULE_NAME, __FUNCTION__, publish_url_.c_str());
  rtmp_handler_ = srs_rtmp_create(publish_url_.c_str());
  if (!rtmp_handler_) {
    log(commons::LOG_ERROR, "%s failed to create rtmp protocol stack", MODULE_NAME);
    return -ERR_FAILED;
  }
  log(commons::LOG_INFO, "%s rtmp handler created", MODULE_NAME);

  int ret;
  do {
    ret = srs_rtmp_set_timeout(rtmp_handler_, SRS_RECV_TIMEOUT_MS, SRS_SEND_TIMEOUT_MS);
    if (ret != 0) {
      log(commons::LOG_ERROR, "%s set timeout failed: %d", MODULE_NAME, ret);
      // no need to break
    }
    log(commons::LOG_INFO, "%s set timeout (recv: %d ms, send: %d ms) success", MODULE_NAME,
        SRS_RECV_TIMEOUT_MS, SRS_SEND_TIMEOUT_MS);

    ret = srs_rtmp_handshake(rtmp_handler_);
    if (ret != 0) {
      log(commons::LOG_ERROR, "%s simple handshake failed: %d", MODULE_NAME, ret);
      break;
    }
    log(commons::LOG_INFO, "%s simple handshake success", MODULE_NAME);

    ret = srs_rtmp_connect_app(rtmp_handler_);
    if (ret != 0) {
      log(commons::LOG_ERROR, "%s connect vhost/app failed: %d", MODULE_NAME, ret);
      break;
    }
    log(commons::LOG_INFO, "%s connect vhost/app success", MODULE_NAME);

    ret = srs_rtmp_publish_stream(rtmp_handler_);
    if (ret != 0) {
      log(commons::LOG_ERROR, "%s publish stream failed: %d", MODULE_NAME, ret);
      break;
    }
    log(commons::LOG_INFO, "%s publish stream success", MODULE_NAME);
  } while (0);

  if (ret != 0) {
    // TODO(Haonong Yu): 2020/6/9 error code mapping
    srs_rtmp_destroy(rtmp_handler_);
    rtmp_handler_ = nullptr;
    return -ret;
  }

  log(commons::LOG_INFO, "%s %s done", MODULE_NAME, __FUNCTION__);
  return -ERR_OK;
}

int RtmpSenderImpl::SendMetaData() {
  std::unordered_map<std::string, srs_amf0_t> properties;

  if (rtmp_config_.framerate > 0) {
    properties["videocodecid"] = srs_amf0_create_number(7);
    properties["width"] = srs_amf0_create_number(rtmp_config_.video_width);
    properties["height"] = srs_amf0_create_number(rtmp_config_.video_height);
    properties["framerate"] = srs_amf0_create_number(rtmp_config_.framerate);
    properties["videodatarate"] = srs_amf0_create_number(rtmp_config_.video_bitrate);
  }

  properties["audiocodecid"] = srs_amf0_create_number(10);
  properties["audiosamplerate"] = srs_amf0_create_number(rtmp_config_.audio_sample_rate_hz);
  properties["audiosamplesize"] = srs_amf0_create_number(rtmp_config_.audio_bytes_per_sample * 8);
  properties["stereo"] = srs_amf0_create_boolean(rtmp_config_.audio_numbder_of_channels == 2);
  return srs_rtmp_write_metadata(rtmp_handler_, properties);
}

void RtmpSenderImpl::CloseRtmp() {
  ASSERT_THREAD_IS(rtmp_sender_worker_->getThreadId());
  log(commons::LOG_INFO, "%s %s", MODULE_NAME, __FUNCTION__);
  srs_rtmp_destroy(rtmp_handler_);
  rtmp_handler_ = nullptr;
  log(commons::LOG_INFO, "%s %s done", MODULE_NAME, __FUNCTION__);
}

int RtmpSenderImpl::SendAudioFrame(const uint8_t* data, size_t size, int64_t pts_ms) {
  if (!is_started_ ||
      is_stopping_) {  // speculatively avoid memory footprint and async task scheduling
    return -ERR_INVALID_STATE;
  }

  struct RtmpMediaFrame audio_frame;
  audio_frame.is_video_frame = false;
  uint8_t* copy_data = reinterpret_cast<uint8_t*>(malloc(size * sizeof(uint8_t)));
  memcpy(copy_data, data, size);
  audio_frame.data = copy_data;
  audio_frame.size = size;
  audio_frame.pts_ms = static_cast<uint32_t>(pts_ms);
  streaming_buffer_->PushFrame(audio_frame);
  return SendPacketAsync();
}

int RtmpSenderImpl::SendVideoFrame(const uint8_t* data, size_t size, int64_t timestamp_ms,
                                   bool is_key) {
  // speculatively avoid memory footprint and async task scheduling

  if (!is_started_ || is_stopping_ || data == nullptr || size == 0) {
    return -ERR_INVALID_STATE;
  }

  struct RtmpMediaFrame video_frame;
  video_frame.is_video_frame = true;
  uint8_t* copy_data = static_cast<uint8_t*>(malloc(size * sizeof(uint8_t)));
  memcpy(copy_data, data, size);
  video_frame.data = copy_data;
  video_frame.size = size;
  video_frame.pts_ms = static_cast<uint32_t>(timestamp_ms);
  video_frame.is_key_frame = is_key;
  streaming_buffer_->PushFrame(video_frame);
  return SendPacketAsync();
}

int RtmpSenderImpl::SendPacketAsync() {
  if (rtmp_sending_queue_ && !is_sending_.exchange(true)) {
    rtmp_sending_queue_->async_call([this] { TrySendPacket(); });
  }
  return 0;
}

int RtmpSenderImpl::TrySendPacket() {
  ASSERT_THREAD_IS(rtmp_sender_worker_->getThreadId());
  if (!is_started_ || !rtmp_handler_ || is_reconnecting_) {
    // Will start sending packet once stream is opened
    is_sending_.store(false);
    return -ERR_INVALID_STATE;
  }

  if (streaming_buffer_->FrameListSize() <= 0) {
    log(commons::LOG_INFO, "%s %s: no frame ready for sending", MODULE_NAME, __FUNCTION__);
    is_sending_.store(false);
    return -ERR_INVALID_STATE;
  }

  int ret = -ERR_OK;
  RtmpMediaFrame frame;
  streaming_buffer_->PopFirstFrame(frame);
  int srs_err = DoSendPacket(frame);
  if (srs_err != ERROR_SUCCESS) {
    log(commons::LOG_ERROR, "%s %s: failed to send %s frame, key: %d err: %d", MODULE_NAME,
        __FUNCTION__, frame.is_video_frame ? "video" : "audio", frame.is_key_frame, srs_err);

    // Removes trailing P frames if failed to send a key video frame
    if (frame.is_video_frame && frame.is_key_frame) {
      streaming_buffer_->RemoveTrailingPFrames();
    }

    // Reconnects if needed
    if (srs_err == ERROR_SOCKET_CLOSED) {
      log(commons::LOG_WARN, "%s %s: socket closed, reconnecting", MODULE_NAME, __FUNCTION__);
      CloseRtmp();
      ReconnectRtmp();
    }

    is_sending_.store(false);
    ret = -ERR_INVALID_STATE;
  } else {
    UpdateStateAndNotifyIfNeeded(kSocketStateSending);

    // schedule task of sending next packet, leaving room for other kinds of sender tasks.
    if (streaming_buffer_->FrameListSize() > 0 && rtmp_sending_queue_) {
      rtmp_sending_queue_->async_call([this] { TrySendPacket(); });
    } else {
      is_sending_.store(false);
    }
  }
  free(frame.data);
  return ret;
}

// return srs error code
int RtmpSenderImpl::DoSendPacket(const RtmpMediaFrame& frame) {
  ASSERT_THREAD_IS(rtmp_sender_worker_->getThreadId());

  // RTMP timestamp should start with 0
  if (relative_timestamps_ == 0) {
    relative_timestamps_ = frame.pts_ms;
    log(commons::LOG_INFO, "%s %s: set relative time base as %u", MODULE_NAME, __FUNCTION__,
        relative_timestamps_);
  }
  uint32_t send_pts_ms =
      frame.pts_ms > relative_timestamps_ ? frame.pts_ms - relative_timestamps_ : 0;

  // Log send start
  int64_t start_ms = 0;
  if (NeedLogStream()) {
    log(commons::LOG_DEBUG, "%s [%d] start sending %s frame, pts: %u size: %zu key: %d",
        MODULE_NAME, send_frame_cnt_, frame.is_video_frame ? "video" : "audio", send_pts_ms,
        frame.size, frame.is_key_frame);
    start_ms = ::rtc::TimeMillis();
  }

  // Send audio/video frame
  int ret = 0;
  if (frame.is_video_frame) {  // video frame
    if (last_statistic_data_) {
      last_statistic_data_->last_send_video_frame_count.fetch_add(1);
      last_statistic_data_->last_send_video_bytes.fetch_add(frame.size);
    }
    if (frame.is_key_frame) {
      webrtc::ParserH264 h264_parser;

      // add jira/NMS-4374
      if (h264_parser.ParseHead(frame.data, static_cast<int>(frame.size)) == 0) {
        uint32_t ww = h264_parser.GetWidth();
        uint32_t hh = h264_parser.GetHeight();
        int width_t = static_cast<int>(ww);
        int height_t = static_cast<int>(hh);
        if (last_video_width_ != width_t || last_video_height_ != height_t) {
          rtmp_config_.video_width = last_video_width_ = width_t;
          rtmp_config_.video_height = last_video_height_ = height_t;
          int tmp = SendMetaData();
          if (tmp != 0) {
            log(commons::LOG_WARN, "%s failed to send meta data: %d", MODULE_NAME, tmp);
            // no need to break
          } else {
            log(commons::LOG_INFO, "%s send meta data success", MODULE_NAME);
          }
        }
      }
    }
    ret = srs_h264_write_raw_frames(rtmp_handler_, reinterpret_cast<char*>(frame.data), frame.size,
                                    send_pts_ms, send_pts_ms);
    if (srs_h264_is_duplicated_sps_error(ret) || srs_h264_is_duplicated_pps_error(ret)) {
      log(commons::LOG_DEBUG, "%s %s ignore duplicated sps/pps", MODULE_NAME, __FUNCTION__);
      // We will not send the duplicated sps/pps packet, but I-frame packet will still be sent.
      ret = 0;
    }
    if (srs_h264_is_dvbsp_error(ret)) {
      log(commons::LOG_DEBUG, "%s %s drop video frame before sps/pps", MODULE_NAME, __FUNCTION__);
      ret = 0;
    }
  } else {                  // audio frame
    int sound_format = 10;  // audio data format default to AAC
    int sound_rate = 0;     // audio sample rate default to 44.1 kHz
    switch (rtmp_config_.audio_sample_rate_hz) {
      case 11025:
        sound_rate = 1;  // 11kHz
        break;
      case 22050:
        sound_rate = 2;  // 22kHz
        break;
      case 44100:
        sound_rate = 3;  // 44.1kHz
        break;
    }
    int sound_size = 1;  // audio sample size default to 16-bit
    switch (rtmp_config_.audio_bytes_per_sample) {
      case 1:
        sound_size = 0;  // 8-bit samples
        break;
      case 2:
        sound_size = 1;  // 16-bit samples
        break;
    }
    int sound_type = 1;  // audio type default to stereo
    switch (rtmp_config_.audio_numbder_of_channels) {
      case 1:
        sound_type = 0;  // Mono sound
        break;
      case 2:
        sound_type = 1;  // Stereo sound
        break;
    }
    if (last_statistic_data_) {
      last_statistic_data_->last_send_audio_bytes.fetch_add(frame.size);
    }
    ret = srs_audio_write_raw_frame(rtmp_handler_, sound_format, sound_rate, sound_size, sound_type,
                                    reinterpret_cast<char*>(frame.data), frame.size, send_pts_ms);
  }

  // Log send end
  if (NeedLogStream()) {
    static uint32_t last_send_pts_ms = 0;
    log(commons::LOG_DEBUG,
        "%s [%d] finish sending %s frame, pts: %u (delta: %d), size: %zu, key: %d,"
        " ret: %d spend %" PRId64 " ms",
        MODULE_NAME, send_frame_cnt_, frame.is_video_frame ? "video" : "audio", send_pts_ms,
        send_pts_ms - last_send_pts_ms, frame.size, frame.is_key_frame, ret,
        ::rtc::TimeMillis() - start_ms);
    last_send_pts_ms = send_pts_ms;
  }

  send_frame_cnt_++;
  return ret;
}

void RtmpSenderImpl::UpdateStateAndNotifyIfNeeded(RtmpSocketState state) {
  ASSERT_THREAD_IS(rtmp_sender_worker_->getThreadId());
  if (current_socket_state_ != state) {
    current_socket_state_ = state;
    sender_observers_->Call(
        [this](auto callback) { callback->OnRtmpSocketStateChanged(current_socket_state_); });
  }
}

int RtmpSenderImpl::WriteFlvFile(const char* filename, bool has_audio, bool has_video) {
  log(commons::LOG_INFO, "%s write flv, file:%s audio:%d video:%d", MODULE_NAME, filename,
      has_audio, has_video);
  // TODO(Haonong Yu): 2020/3/13 cache write request if not started yet. And thread safety.
  //  std::string file_name_str = filename;
  return rtmp_sender_worker_->async_call(LOCATION_HERE, [=] {
    log(commons::LOG_INFO, "%s write flv file, stream opened: %d", MODULE_NAME,
        rtmp_handler_ != nullptr);
    // TODO(Haonong Yu): 2020/6/9 flv write using srs-librtmp
    //    flv_file_open(file_name_str.c_str());
    //    write_flv_header(has_audio, has_video);
    return 0;
  });
}

void RtmpSenderImpl::WriteFlvClose() {
  log(commons::LOG_INFO, "%s write flv close", MODULE_NAME);
  rtmp_sender_worker_->async_call(LOCATION_HERE, [] {
    // TODO(Haonong Yu): 2020/6/9 flv write using srs-librtmp
    //    flv_file_close();
    return 0;
  });
}

int RtmpSenderImpl::RegisterRtmpSenderObserver(IRtmpSenderObserver* observer) {
  log(commons::LOG_INFO, "%s %s %p", MODULE_NAME, __FUNCTION__, observer);
  if (!observer) {
    return -ERR_INVALID_ARGUMENT;
  }
  return sender_observers_->Register(observer);
}

int RtmpSenderImpl::UnregisterRtmpSenderObserver(IRtmpSenderObserver* observer) {
  log(commons::LOG_INFO, "%s %s %p", MODULE_NAME, __FUNCTION__, observer);
  if (!observer) {
    return -ERR_INVALID_ARGUMENT;
  }
  return sender_observers_->Unregister(observer);
}

void RtmpSenderImpl::StatisticTask() {
  if (is_started_ && last_statistic_data_) {
    if (last_statistic_data_->last_check_time <= 0) {
      last_statistic_data_->last_check_time = commons::now_us();
      return;
    }

    const int S_US_INTERVAL = 1000000;
    auto now = commons::now_us();
    double dur_s = static_cast<double>(now - last_statistic_data_->last_check_time) / S_US_INTERVAL;

    RtmpSenderStatisticsInfo s_info = {0};
    s_info.video_width = last_video_width_;
    s_info.video_height = last_video_height_;
    s_info.video_bitrate =
        static_cast<uint64_t>(
            static_cast<double>(last_statistic_data_->last_send_video_bytes.load()) *
            sizeof(uint8_t) / dur_s) *
        8;
    s_info.audio_bitrate =
        static_cast<uint64_t>(
            static_cast<double>(last_statistic_data_->last_send_audio_bytes.load()) *
            sizeof(uint8_t) / dur_s) *
        8;
    s_info.video_frame_rate = static_cast<uint64_t>(
        static_cast<double>(last_statistic_data_->last_send_video_frame_count.load()) / dur_s);
    last_statistic_data_->Clear();
    last_statistic_data_->last_check_time = now;

    if (streaming_buffer_) {
      auto buffer_statistics = streaming_buffer_->GetStatisticInfo();
      s_info.push_video_frame_cnt = buffer_statistics.push_video_frame_cnt_;
      s_info.push_audio_frame_cnt = buffer_statistics.push_audio_frame_cnt_;
      s_info.skip_video_pframe_cnt = buffer_statistics.skip_video_pframe_cnt_;
      s_info.skip_video_iframe_cnt = buffer_statistics.skip_video_iframe_cnt_;
      s_info.pop_video_frame_cnt = buffer_statistics.pop_video_frame_cnt_;
      s_info.pop_audio_frame_cnt = buffer_statistics.pop_audio_frame_cnt_;
    }

    sender_observers_->Call([=](auto callback) { callback->OnRtmpSenderStatisticsInfo(s_info); });

  } else {
    if (last_statistic_data_) {
      last_statistic_data_->Clear();
    }
  }
}

void RtmpSenderImpl::ReconnectRtmp() {
  ASSERT_THREAD_IS(rtmp_sender_worker_->getThreadId());

  if (!is_started_ || is_stopping_) {
    CleanUpReconnection();
    log(commons::LOG_INFO, "%s no more reconnecting since streaming is stopped", MODULE_NAME);
    return;
  }

  is_reconnecting_ = true;
  ++reconnect_count_current_;
  log(commons::LOG_WARN, "%s %s curr: %d max: %d", MODULE_NAME, __FUNCTION__,
      reconnect_count_current_, reconnect_count_max_);
  UpdateStateAndNotifyIfNeeded(kSocketStateReconnecting);

  int ret = OpenRtmp();
  log(commons::LOG_INFO, "%s reconnect rtmp ret: %d, url: %s", MODULE_NAME, ret,
      publish_url_.c_str());
  log(commons::LOG_INFO, "reconnect_count_: %d, reconnect_max: %d", reconnect_count_current_,
      reconnect_count_max_);

  if (ret == 0) {
    CleanUpReconnection();
    streaming_buffer_->CleanUp();
    UpdateStateAndNotifyIfNeeded(kSocketStateReconnected);
    is_sending_.store(true);
    TrySendPacket();
  } else if (reconnect_count_current_ >= reconnect_count_max_) {
    CleanUpReconnection();
    UpdateStateAndNotifyIfNeeded(kSocketStateReconnectTimeout);
  } else {  // retry reconnection
    if (!reconnect_timer_) {
      reconnect_timer_.reset(
          rtmp_sender_worker_->createTimer(std::bind(&RtmpSenderImpl::ReconnectRtmp, this), 1000));
    }
  }
}

void RtmpSenderImpl::CleanUpReconnection() {
  is_reconnecting_ = false;
  reconnect_count_current_ = 0;
  reconnect_timer_.reset();
}

void RtmpSenderImpl::OnRtmpStreamingBufferState(RtmpStreamingBufferState status) {
  sender_observers_->Call(
      [this, status](auto callback) { callback->OnRtmpStreamingBufferState(status); });
}

}  // namespace rtc
}  // namespace agora
